package com.atguigu.mapreduce.solution;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.shaded.org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class SolutionReducer extends Reducer<Text, OrderBean, OrderBean, NullWritable> {
    private final List<OrderBean> orderList = new ArrayList<>();
    private final List<OrderBean> userList = new ArrayList<>();

    @Override
    protected void reduce(Text key, Iterable<OrderBean> values, Reducer<Text, OrderBean, OrderBean, NullWritable>.Context context) throws IOException, InterruptedException {
        int orderIdx = 0;
        int userIdx = 0;
        for (OrderBean value : values) {
            OrderBean result;
            if (StringUtils.isEmpty(value.getEmail())) {
                while (orderIdx >= orderList.size()) {
                    orderList.add(new OrderBean());
                }
                result = orderList.get(orderIdx);
                orderIdx++;
            } else {
                while (userIdx >= userList.size()) {
                    userList.add(new OrderBean());
                }
                result = userList.get(userIdx);
                userIdx++;
            }
            result.setOrderId(value.getOrderId());
            result.setUserId(value.getUserId());
            result.setAmount(value.getAmount());
            result.setEmail(value.getEmail());
        }

        for (int i = 0; i < orderIdx; i++) {
            for (int j = 0; j < userIdx; j++) {
                OrderBean order = orderList.get(i);
                OrderBean user = userList.get(j);
                order.setEmail(user.getEmail());
                context.write(order, NullWritable.get());
            }
        }
    }
}
